package interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author Jeremy Zheng
 * @Date 2021/4/7 14:59
 * @Version 1.0
 */
public class LogETLInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        //获取event数据
        byte[] body = event.getBody();
        //转换成字符串格式，针对字符串进行操作
        // 421 public String(byte bytes[], int offset, int length, StringcharsetName)
        String log = new String(body, Charset.forName("utf-8"));
        //判断是启动日志（json）还是事件日志（timestamp|json）：包含“start”的是启动日志，否则是事件日志
        if (log.contains("start")){
            if (LogUtils.checkStart(log)){ //经过检查判断是start日志，返回，不是的被过滤掉
                return event;
            }
        }else {
            if (LogUtils.checkEvent(log)){ //经过检查判断是事件日志，返回，不是的脏数据被过滤掉
                return event;
            }
        }
        return null; //不是启动日志也不是事件日志，过滤掉，返回空
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        ArrayList<Event> eventList = new ArrayList<>();
        for (Event event : events) {
            Event intercept = intercept(event);
            if (intercept != null){
                eventList.add(intercept);
            }
        }
        return eventList;
    }

    @Override
    public void close() {

    }

    //给自定义ETL拦截器一个静态内部类Builder
    //因为在Flume中使用的时候不能new一个LogETLInterceptor()对象，也就无法调用其处理event的方法
    //此处的静态内部类Builder让我们直接使用类名点方法去完成实例对象的创建
    public static class Builder implements Interceptor.Builder{
        @Override
        public Interceptor build() {
            return new LogETLInterceptor();
        }
        @Override
        public void configure(Context context) {
        }

    }

}
